package com.navinfo.platform.flink.cdc.streaming.functions;

import com.alibaba.fastjson.JSONObject;
import com.navinfo.platform.flink.cdc.dao.DeleteDao;
import com.navinfo.platform.flink.cdc.dao.UpsertDao;
import com.navinfo.platform.flink.cdc.util.DBUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;

/**
 * 通过一个ProcessFunction进行所有业务处理
 * 所有表共享一个事件源,为了保证state中的消费位置和消费顺序,并发只能设置为1
 * 即,处理完上一个,再处理下一个,完全不能有任何并行
 * 所以即使将这个ProcessFunction拆分成多个,也不能增加并行度
 * 所以,这个Function中就做了所有的业务处理,没必要拆分成多个步骤了
 * 但插入日志表这个操作,则给到下一个ProcessFunction来进行处理
 */
public class SynchronizeFunction extends ProcessFunction<String, JSONObject> {
    private static final Logger logger = LoggerFactory.getLogger(SynchronizeFunction.class);
    private final ParameterTool config;

    UpsertDao upsertDao = new UpsertDao();
    DeleteDao deleteDao = new DeleteDao();

    public SynchronizeFunction(ParameterTool config) {
        this.config = config;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //本类是第一次使用数据库连接,在open中初始化数据库连接池
        DBUtils.init(config);
    }

    /**
     * 处理一个消息
     * 这里有几种方式:
     * 1)反复重试方式
     * 这里采用while(true)的方式,即一个消息如果处理失败,则反复重试,直到处理成功为止
     * 优点: 如果sink数据库挂了,或关跛的接口挂了,或高德的http请求无法接通,则会在上述情况回复后自动恢复
     * 缺点: 整个同步过程会被卡住,而且如果有业务代码异常,则也会卡住
     * 2)不重试,加异常处理
     * 一旦抛出异常,则抛弃掉此消息,继续处理下一个消息
     * 优点:不会造成处理流程卡住
     * 缺点: 如果sink数据库,关跛接口,高德http接口短期不可用,则会造成数据不一致
     * 3)什么也不做
     * 如果用一个普通的自定义Source,抛异常后会导致整个flink job退出,这样集群就知道这个任务挂了
     * 但对于cdc,一旦抛出异常,则debezium会停止服务,并不再接收后续消息
     * 这应该算cdc的一个小问题,就是debezium停止服务后,整个flink job并没有退出
     * 这可能导致一个问题,就是一旦抛异常,flink job还在,但是却不工作了
     * <p>
     */
    @Override
    public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        logger.info("新的数据:{}", value);
        //1.解析json字符串
        JSONObject message = JSONObject.parseObject(value);
        JSONObject after = message.getJSONObject("after");
        JSONObject before = message.getJSONObject("before");
        String tableName = message.getJSONObject("source").getString("table");
        try {
            //2.根据不同类型的操作,调用不同的方法
            String op = message.getString("op");
            switch (op) {
                // 2.0快照时处理
                case "r": //只读
                case "c": //新增
                case "u": //更新
                    //4.执行同步操作
                    this.upsert(after, tableName);
                    break;
                case "d": //删除
                    this.delete(before, tableName);
                    break;
            }
        } catch (Exception e) {
            logger.error("处理消息过程中出现异常", e);
        }
        //6.下一个Function
        out.collect(message);
    }

    /**
     * upsert操作,所有表的新增和更新都是由这个方法处理
     */
    private void upsert(JSONObject after, String tableName) throws SQLException {
        switch (tableName) {
            case "hy_fault_detail":
                upsertDao.upsertHyFaultDetail(after);
                break;
        }
    }

    /**
     * delete操作(对sink中的表,进行逻辑删除)
     */
    private void delete(JSONObject before, String tableName) throws SQLException {
        switch (tableName) {
            case "hy_fault_detail":
                deleteDao.deleteHyFaultDetail(before);
                break;
        }
    }
}
